package cn.edu.xmu.datamining.tangzk.mralgos.kmeans;

import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class KMeansAssignMapper extends
		Mapper<LongWritable, Text, IntWritable, Text> {

	private static final Logger LOG = LoggerFactory
			.getLogger(KMeansMapper.class);
	private List<ClusterCentroidWritable> clusterCenters = null;

	protected void setup(
			org.apache.hadoop.mapreduce.Mapper<LongWritable, Text, IntWritable, Text>.Context context)
			throws java.io.IOException, InterruptedException {
		Configuration conf = context.getConfiguration();
		clusterCenters = new ArrayList<ClusterCentroidWritable>();
		int badRecords = KMeansMapper.loadClustersFromDistributedCache(
				clusterCenters, conf);
		context.getCounter(KMeansDriver.USER_DEFINED_GROUP,
				KMeansDriver.CNT_BAD_CLUSTER_RECOD).increment(badRecords);
	};

	protected void map(
			LongWritable key,
			Text value,
			org.apache.hadoop.mapreduce.Mapper<LongWritable, Text, IntWritable, Text>.Context context)
			throws java.io.IOException, InterruptedException {
		String val = value.toString();
		SampleWritable sample = SampleWritable.parseSample(val);
		ClusterCentroidWritable closestCentroid = sample
				.getClosestCluster(clusterCenters);
		if (closestCentroid == null) {
			LOG.error("can't find cluster({}): {}", closestCentroid,
					value.toString());
			context.getCounter(KMeansDriver.USER_DEFINED_GROUP,
					KMeansDriver.CNT_BAD_RECORDS).increment(1);
			return;
		}
		context.write(new IntWritable(closestCentroid.getC()), value);

	};

}
